/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.example.ordermessage;

import org.apache.rocketmq.client.exception.MQBrokerException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.MessageQueueSelector;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageQueue;
import org.apache.rocketmq.example.constant.RocketMqConstant;
import org.apache.rocketmq.remoting.common.RemotingHelper;
import org.apache.rocketmq.remoting.exception.RemotingException;

import java.io.UnsupportedEncodingException;
import java.util.List;

/**
 * https://blog.csdn.net/maimai1993/article/details/125033486
 */
public class Producer {
    /**
     * 局部顺序消费：通常在实际应用中，我们需要将同一个订单号的相关操作，
     * 按照规则（可以是hash或取模等）发送到同一个queue上（使用MessageQueueSelector ），
     * 然后消费者实例，使用顺序消费模式消费消息（使用MessageListenerOrderly）。
     *
     * @param args
     * @throws UnsupportedEncodingException
     */
    public static void main(String[] args) throws UnsupportedEncodingException {
        try {
            DefaultMQProducer producer = new DefaultMQProducer("please_rename_unique_group_name");
            producer.setNamesrvAddr(RocketMqConstant.NAME_SRV_ADDR);
            producer.start();
            while (true) {
                for (int i = 0; i < 10; i++) {
                    int orderId = i % 10;
                    for (int j = 0; j < 10; j++) {
                        Message msg = new Message("OrderTopicTest", "order_" + orderId, "KEY" + orderId, ("order_" + orderId + " step " + j).getBytes(RemotingHelper.DEFAULT_CHARSET));
                        SendResult sendResult = producer.send(msg, new MessageQueueSelector() {
                            /**
                             *
                             * @param mqs
                             * @param msg
                             * @param arg  也就是orderId
                             * @return
                             */
                            @Override
                            public MessageQueue select(List<MessageQueue> mqs, Message msg, Object arg) {

                                Integer id = (Integer) arg;
                                int index = id % mqs.size();
                                return mqs.get(index);
                            }
                        }, orderId); //此时的orderId就是org.apache.rocketmq.client.producer.MessageQueueSelector.select 这个方法中的参数
                        System.out.println(new String(msg.getBody()));
                        System.out.printf("%s%n", sendResult);
                    }

                }
            }


//            producer.shutdown();
        } catch (MQClientException | RemotingException | MQBrokerException | InterruptedException e) {
            e.printStackTrace();
        }
    }
}
